package com.sqk.task0708;
import java.io.IOException;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class DxCountPvMain {

	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Job job1 = Job.getInstance();
		job1.setJarByClass(DxCountPvMain.class);
		job1.setMapperClass(DxCountPvMapper.class);
		job1.setReducerClass(DxCountPvReducer.class);
		
		job1.setMapOutputKeyClass(Text.class);
		job1.setMapOutputValueClass(IntWritable.class);
		job1.setOutputKeyClass(Text.class);
		job1.setOutputValueClass(IntWritable.class);
		
		FileInputFormat.setInputPaths(job1, new Path("F:\\SparkTask\\DxProBehave"));
		FileOutputFormat.setOutputPath(job1, new Path("F:\\SparkTask\\DxFileTopPv7"));
		if(job1.waitForCompletion(true)) {
			Job job2 = Job.getInstance();
			job2.setJarByClass(DxCountPvMain.class);
			job2.setMapperClass(DxCountUvPvMapper.class);
			job2.setReducerClass(DxCountUvPvReducer.class);
			
			job2.setMapOutputKeyClass(Text.class);
			job2.setMapOutputValueClass(Text.class);
			job2.setOutputKeyClass(NullWritable.class);
			job2.setOutputValueClass(Text.class);
			
			FileInputFormat.addInputPath(job2, new Path("F:\\SparkTask\\DxFileTopPv7"));
		    FileOutputFormat.setOutputPath(job2,
		    	      new Path("F:\\SparkTask\\DxFileTopPv8"));
		    System.exit(job2.waitForCompletion(true) ? 0 : 1);
		}

	}

}
